package cn.doitedu.eagle.queryservice;

import cn.doitedu.eagle.beans.*;
import cn.doitedu.eagle.names.MetricNames;
import cn.doitedu.eagle.ruleutils.ClickHouseUtil;
import cn.doitedu.eagle.ruleutils.RuleParamUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.metrics.Counter;

import java.sql.*;
import java.util.*;
import java.util.Date;

/**
 * @author 涛哥
 * @nick_name "deep as the sea"
 * @contact qq:657270652 wx:doit_edu
 * @site www.doitedu.cn
 * @date 2021-03-15
 * @desc 用户行为序列查询服务
 */
@Slf4j
public class UserActionSequenceQueryServiceImpl implements UserActionSequenceQueryService {
    // 事件beans缓存
    ListState<LogBean> eventState;
    // 缓存管理器
    FlinkStateBufferManager flinkStateBufferManager;
    // clickHouse连接
    Connection conn;
    // ck查询metric计数器
    Counter ckQueryCounter;
    // 缓存有效次数metric计数器
    Counter bufferAvailCounter;


    public UserActionSequenceQueryServiceImpl(RuntimeContext runtimeContext,ListState<LogBean> eventState,
                                              FlinkStateBufferManager flinkStateBufferManager) throws Exception {
        this.eventState = eventState;
        this.flinkStateBufferManager = flinkStateBufferManager;

        conn = ClickHouseUtil.getClickHouseConnection();

        ckQueryCounter = runtimeContext.getMetricGroup().counter(MetricNames.CK_QUERY_CNTS);
        bufferAvailCounter = runtimeContext.getMetricGroup().counter(MetricNames.BUFFER_AVAIL_CNTS);

    }


    /**
     * 用户行为序列条件查询服务入口方法
     *
     * @param ruleParam 规则参数
     * @return 最大步骤号
     * @throws Exception
     */
    public boolean findActionSequenceMaxStep(RuleParam ruleParam) throws Exception {

        long curTime = System.currentTimeMillis();
        String deviceId = ruleParam.getDeviceId();
        List<RuleParamEvent> seqParam = ruleParam.getEventSequenceParamList();
        int stepAmount = seqParam.size();

        RuleParamEvent eventParam = ruleParam.getEventSequenceParamList().get(0);
        // 这里是 年月日时分秒格式
        String timeRangeStart = eventParam.getTimeRangeStart();
        String timeRangeEnd = eventParam.getTimeRangeEnd();

        // 转成long格式
        long queryStartTime = DateUtils.parseDate(timeRangeStart, "yyyy-MM-dd HH:mm:ss").getTime();
        // 如果end是-1，则表示要计算到当前；
        long queryEndTime = "-1".equals(timeRangeEnd) ? curTime : DateUtils.parseDate(timeRangeEnd, "yyyy-MM-dd HH:mm:ss").getTime();
        // 远近分界点（当前时间-15分钟的前1小时对小时取整）
        long splitPoint = DateUtils.ceiling(DateUtils.addMinutes(new Date(curTime), -75), Calendar.HOUR).getTime();


        /*
         * 先从缓存中查询
         *
         * **/
        String bufferKey = RuleParamUtil.ruleEventParamList2Str(ruleParam.getEventSequenceParamList(), deviceId);

        BufferResult bufferResult = flinkStateBufferManager.getSeqBuffer(bufferKey, queryStartTime, queryEndTime, splitPoint);
        BufferAvailLevel availLevel = bufferResult.getAvailLevel();

        // buffer有效，查询截短后的近期
        if (availLevel == BufferAvailLevel.FAR_NEAR_PARTIAL) {
            bufferAvailCounter.inc();
            log.error("规则：{},序列缓存FAR_NEAR_PARTIAL有效：{} ,查询的的deviceId：{},缓存返回：{},序列总步骤：{},查询start: {},查询end: {},buff_start: {},buff_end: {},分界点: {},",ruleParam.getRuleId(),
                    bufferKey, deviceId, bufferResult.getBufferValue(), stepAmount, queryStartTime, queryEndTime, bufferResult.getBufferStartTime(), bufferResult.getBufferEndTime(), splitPoint);

            int bufferValue = bufferResult.getBufferValue();
            if (bufferValue >= stepAmount) {
                log.error("规则：{},序列缓存FAR_NEAR_PARTIAL命中：{} ,查询的的deviceId：{},缓存返回：{},序列总步骤：{},查询start: {},查询end: {},buff_start: {},buff_end: {},分界点: {},规则：{}",ruleParam.getRuleId(),
                        bufferKey, deviceId, bufferResult.getBufferValue(), stepAmount, queryStartTime, queryEndTime, bufferResult.getBufferStartTime(), bufferResult.getBufferEndTime(), splitPoint);
                return true;
            }
            // 截短序列
            List<RuleParamEvent> subSeq = seqParam.subList(bufferValue, stepAmount);

            // 查询state
            int stepsNear = findActionSequenceNear(eventState.get(), subSeq, bufferResult.getShiftStartTime(), bufferResult.getShiftEndTime());

            // 更新缓存
            flinkStateBufferManager.updateSeqBuffer(bufferKey, bufferValue + stepsNear, queryStartTime, queryEndTime);

            // 如果条件以满足，直接返回true，否则直接返回false
            return bufferValue + stepsNear >= stepAmount;
        }
        // buffer无效，则进行正常查询
        else {
            int stepFar = 0;
            int stepNear = 0;
            int stepRes = 0;
            // 只查ck
            if (queryEndTime <= splitPoint) {
                stepRes = findActionSequenceFar(ruleParam, queryStartTime, queryEndTime);
                ckQueryCounter.inc();

                // 更新缓存
                flinkStateBufferManager.updateSeqBuffer(bufferKey,stepRes,queryStartTime,queryEndTime);

                // 返回结果
                return stepFar >= stepAmount;
            }
            // 双路查询
            else if (queryStartTime < splitPoint && queryEndTime > splitPoint) {
                stepFar = findActionSequenceFar(ruleParam, queryStartTime, splitPoint);
                ckQueryCounter.inc();
                stepNear = findActionSequenceNear(eventState.get(), seqParam, splitPoint, queryEndTime);
                stepRes = stepFar + stepNear;

                // 更新缓存
                flinkStateBufferManager.updateSeqBuffer(bufferKey,stepRes,queryStartTime,queryEndTime);

                // 返回结果
                return stepRes >= stepAmount;
            }
            // 只查state
            else {
                stepRes = findActionSequenceNear(eventState.get(), seqParam, queryStartTime, queryEndTime);

                // 更新缓存
                flinkStateBufferManager.updateSeqBuffer(bufferKey,stepRes,queryStartTime,queryEndTime);

                // 返回结果
                return stepRes >= stepAmount;
            }
        }

    }


    /**
     * 内部工具方法：从外部存储中查询最大满足的序列步骤
     *
     * @param ruleParam
     * @return 最大完成步骤
     */
    private int findActionSequenceFar(RuleParam ruleParam, long startTime, long endTime) throws Exception {
        String deviceId = ruleParam.getDeviceId();
        int stepAmt = ruleParam.getEventSequenceParamList().size();
        String buffKey = RuleParamUtil.ruleEventParamList2Str(ruleParam.getEventSequenceParamList(), deviceId);


        String sql = ruleParam.getEventSeqQuerySql();
        PreparedStatement st = conn.prepareStatement(sql);

        // 设置sql条件 deviceid
        st.setString(1, deviceId);
        st.setLong(2, startTime);
        st.setLong(3, endTime);


        long qs = System.currentTimeMillis();
        ResultSet resultSet = st.executeQuery();
        //ResultSetMetaData rsmd = resultSet.getMetaData();

        // 查询结果：（条件步骤总数，最大完成步骤）
        /**
         * ┌─deviceId─────┬─is_match3─┬─is_match2─┬─is_match1─┐
         * │ rIzRTdlM2vjk │         0 │         0 │         1 │
         * └──────────────┴───────────┴───────────┴───────────┘
         */
        int maxStep = 0;
        if (resultSet.next()) {
            for (int i = 2; i <= stepAmt + 1; i++) {
                int step = resultSet.getInt(i);
                if (step == 1) {
                    maxStep = stepAmt - i + 2;
                    log.debug("找到最大匹配步骤:{} ",maxStep);
                    break;
                }
            }
        }
        long qe = System.currentTimeMillis();
        log.warn("规则：{},序列条件ck查询：{} ,deviceId：{},从ck返回：{},序列总步骤：{},queryStart：{},queryEnd：{},耗时：{} 毫秒",ruleParam.getRuleId(),buffKey,deviceId,maxStep,stepAmt,startTime,endTime,qe-qs);

        return maxStep;
    }


    /**
     * 内部工具方法：从state中查询最大满足的序列步骤
     *
     * @param beans    用户的事件列表
     * @param seqParam 规则中定义好的行为序列
     * @return 最大满足到的序列中的步骤
     */
    private int findActionSequenceNear(Iterable<LogBean> beans, List<RuleParamEvent> seqParam, long startTime, long endTime) {

        // 取出条件中涉及到的事件(其他无关事件，不参与后续的运算）
        ArrayList<LogBean> relLogBeans = new ArrayList<>();
        for (RuleParamEvent ruleParamEvent : seqParam) {
            for (LogBean bean : beans) {
                if (isEqual(bean, ruleParamEvent) && bean.getTimeStamp() >= startTime && bean.getTimeStamp() <= endTime) {
                    relLogBeans.add(bean);
                }
            }
        }

        // 对过滤出来的bean进行排序
        Collections.sort(relLogBeans, new Comparator<LogBean>() {
            @Override
            public int compare(LogBean o1, LogBean o2) {
                return Long.compare(o1.getTimeStamp(), o2.getTimeStamp());
            }
        });


        int step = 0;
        // 如果过滤后的list没有匹配事件，直接返回0
        if (relLogBeans.size() < 1) return step;

        // 从排序好的beans中查找规则序列
        long tmpTimestamp = 0;
        for (RuleParamEvent ruleParamEvent : seqParam) {
            boolean find = false;
            for (LogBean relLogBean : relLogBeans) {
                // 如果找到了，并且当前bean的时间戳>=上一个找到的bean的时间戳
                if (ruleParamEvent.getEventId().equals(relLogBean.getEventId()) && isEqual(relLogBean, ruleParamEvent) && relLogBean.getTimeStamp() >= tmpTimestamp) {
                    // 则step++
                    step++;
                    // 且将tmp时间戳记录本次的时间戳
                    tmpTimestamp = relLogBean.getTimeStamp();
                    // 将find标记置为true
                    find = true;
                    // 并且跳出本轮循环
                    break;
                }
            }

            // 如果某个步骤条件，没有找到符合条件的bean，则直接退出外层循环
            if (!find) break;
        }
        return step;
    }

    /**
     * 比较一个日志bean事件对象，是否满足规则条件中定义的事件条件
     *
     * @param bean  日志bean事件
     * @param param 规则中定义的事件条件
     * @return 是否吻合
     */
    private boolean isEqual(LogBean bean, RuleParamEvent param) {
        if (bean.getEventId().equals(param.getEventId())) {
            HashMap<String, String> beanProperties = bean.getProperties();
            Map<String, String> paramProperties = param.getProperties();
            Set<String> paramKeys = paramProperties.keySet();

            // 只要有一个属性条件不满足，则返回false
            for (String paramKey : paramKeys) {
                if (!paramProperties.get(paramKey).equals(beanProperties.get(paramKey))) {
                    return false;
                }
            }
            return true;
        }
        // 如果事件id不相等，直接返回false
        return false;
    }


    // 测试代码
    public static void main(String[] args) throws Exception {

    }

}
